package com.qianfeng.reactor;

//
//                            _ooOoo_
//                           o8888888o
//                           88" . "88
//                           (| -_- |)
//                            O\ = /O
//                        ____/`---'\____
//                      .   ' \\| |// `.
//                       / \\||| : |||// \
//                     / _||||| -:- |||||- \
//                       | | \\\ - /// | |
//                     | \_| ''\---/'' | |
//                      \ .-\__ `-` ___/-. /
//                   ___`. .' /--.--\ `. . __
//                ."" '< `.___\_<|>_/___.' >'"".
//               | | : `- \`.;`\ _ /`;.`/ - ` : | |
//                 \ \ `-. \_ __\ /__ _/ .-` / /
//         ======`-.____`-.___\_____/___.-`____.-'======
//                            `=---='
//
//         .............................................
//                  佛祖镇楼                  BUG辟易
//          佛曰:
//                  写字楼里写字间，写字间里程序员；
//                  程序人员写程序，又拿程序换酒钱。
//                  酒醒只在网上坐，酒醉还来网下眠；
//                  酒醉酒醒日复日，网上网下年复年。
//                  但愿老死电脑间，不愿鞠躬老板前；
//                  奔驰宝马贵者趣，公交自行程序员。
//                  别人笑我忒疯癫，我笑自己命太贱；
//


import reactor.core.publisher.Flux;

import java.util.ArrayList;
import java.util.concurrent.CountDownLatch;

/**
 * Created by Jackiechan on 2022/10/18 9:32
 *
 * @author Jackiechan
 * @version 1.0
 * @since 1.0
 */
public class FluxTest {
    private static CountDownLatch countDownLatch = new CountDownLatch(1);

    public static void main(String[] args) throws InterruptedException {
        //利用数据创建出一个flux对象
        Flux<String> flux = Flux.create(fluxSink -> {
            System.err.println("生产数据了");
            fluxSink.next("wyf")//生产下一个数据
                    .next("dsg")//可以不断的生产数据
                    .next("两个黄鹂鸣翠柳,你还没有女朋友,抽刀断水水更流,你还没有男朋友")
                    .complete();//数据生产完成
        });
        // flux.subscribe();//无参的就是我只订阅,但是不处理数据
        //flux.subscribe(System.err::println);
        //使用Subscriber来订阅
//        flux.subscribe(new Subscriber<String>() {
//            private Subscription subscription;
//
//            /**
//             * 和生产者建立联系,准备问生产要数据
//             * @param subscription
//             */
//            @Override
//            public void onSubscribe(Subscription subscription) {
//                System.err.println("FluxTest中的方法onSubscribe 执行了--->");
//                subscription.request(1);//向上游的生产者要1个数据
//                this.subscription = subscription;
//            }
//
//            /**
//             * 当我们的消费者收到数据的时候执行
//             * @param s
//             */
//            @Override
//            public void onNext(String s) {
//                System.err.println("FluxTest中的方法onNext 执行了--->"+s);
//                //要下一条数据
//                try {
//                    Thread.sleep(5000);
//                } catch (InterruptedException e) {
//                    throw new RuntimeException(e);
//                }
//                subscription.request(1);
//            }
//
//            /**
//             * 发生错误的时候执行
//             * @param throwable
//             */
//            @Override
//            public void onError(Throwable throwable) {
//                System.err.println("FluxTest中的方法onError 执行了--->");
//            }
//
//            /**
//             * 当数据已经处理完成的时候执行
//             */
//            @Override
//            public void onComplete() {
//                System.err.println("FluxTest中的方法onComplete 执行了--->");
//            }
//        });


        Flux<Object> generate = Flux.generate(synchronousSink -> {
            //这个方法只允许调用一次
            synchronousSink.next("老王");
            //synchronousSink.next("老王");
            // synchronousSink.next("老王");
            synchronousSink.complete();
        });
        //generate.subscribe(System.err::println);

        //Flux.just("外面没有人做核酸了").subscribe(System.err::println);
        //Flux.just("外面没有人做核酸了", "做核酸也是应付公事", "尼龙棒破碰了下我的舌尖就结束了", "赵健媳妇密接了,社区说赵健可以随意进出,他媳妇不行").subscribe(System.err::println);

        //Flux.from(Flux.just("外面没有人做核酸了", "做核酸也是应付公事", "尼龙棒破碰了下我的舌尖就结束了", "赵健媳妇密接了,社区说赵健可以随意进出,他媳妇不行")).subscribe(System.err::println);
        //Flux.from(Mono.just("我们有个同事结婚了,秘而不宣,被我侦破了")).subscribe(System.err::println);

        // Flux.fromArray(new String[]{"不要在我面前搞小秘密", "很容易就被我推导出来", "有钱有别墅的杰哥"}).subscribe(System.err::println);

        ArrayList<String> list = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            list.add("老王的金屋" + i);
        }
//        Flux.fromIterable(list).subscribe(System.err::println);


        //Flux.fromStream(list.stream()).subscribe(System.err::println);


        //   Flux.defer(() -> Flux.just("just", "just1", "just2")).subscribe(System.err::println);
        //每个1秒中生成一个数据
//        Flux.interval(Duration.ofMillis(1000)).subscribe(System.err::println);

        //参数1代表3秒后才开始, 第二参数代表每个多少时间生成一条数据
        //  Flux.interval(Duration.ofMillis(3000), Duration.ofMillis(200)).subscribe(System.err::println);
        //参数1 代表从几开始,参数2要几个,包括第一个参数的值
        //Flux.range(10,5).subscribe(System.err::println);

        //=========操作函数=====
        //拼接,必须是前面的生产的数据订阅完成后才会拼接新的数据
        // Flux.just("1","2","3","4").concatWithValues("a","b","c","d").subscribe(System.err::println);
//        Flux.create(fluxSink -> {
//            fluxSink.next("9").next("8").next("7");
//            fluxSink.complete();//结束
//        }).concatWithValues("a","b","c","d").subscribe(System.err::println);

//        Flux.concat( Flux.just("1","2","3","4"), Flux.just("a","b","c","d")).subscribe(System.err::println);

//        Flux.concat(Flux.just(1, 2, 0)
//                        .doOnNext(integer -> System.err.println(100 / integer))//doOnNext在执行next之前先执行的操作
//                , Flux.just(4, 5, 6)).subscribe(new Subscriber<Integer>() {
//            Subscription subscription;
//
//            @Override
//            public void onSubscribe(Subscription s) {
//                s.request(1);
//                subscription = s;
//            }
//
//            @Override
//            public void onNext(Integer integer) {
//                System.err.println(integer);
//                subscription.request(1);
//            }
//
//            @Override
//            public void onError(Throwable t) {
//                t.printStackTrace();
//            }
//
//            @Override
//            public void onComplete() {
//                System.err.println("TestMain中的方法onComplete 执行了");
//            }
//        });

        //前面的数据失败的情况下可以继续执行
//        Flux.concatDelayError(Flux.just(1, 2, 0)
//                        .doOnNext(integer -> System.out.println(100/integer))//doOnNext在执行next之前先执行的操作
//                , Flux.just(4, 5, 6)).subscribe(new Subscriber<Integer>() {
//            Subscription subscription;
//            @Override
//            public void onSubscribe(Subscription s) {
//                s.request(1);
//                subscription = s;
//            }
//
//            @Override
//            public void onNext(Integer integer) {
//                System.err.println(integer);
//                subscription.request(1);
//            }
//
//            @Override
//            public void onError(Throwable t) {
//                t.printStackTrace();
//            }
//
//            @Override
//            public void onComplete() {
//                System.err.println("TestMain中的方法onComplete 执行了");
//            }
//        });
        //只获取第一个生产者的数据
        //Flux.first(Flux.just(1, 2, 0),Flux.just(4, 5, 6)).subscribe(System.err::println);

//        Flux.merge(
//                Flux.interval(Duration.ofMillis(0), Duration.ofMillis(1000)).take(5),
//                        Flux.interval(Duration.ofMillis(500), Duration.ofMillis(1000)).take(5))
//                .subscribe(System.err::println);


        // Flux.mergeComparing(Flux.just(1, 2, 6,8),Flux.just(4, 7, 5,8)).subscribe(System.err::println);
//        Flux.mergeComparing((i1, i2) -> {
//            System.err.println("i1===>" + i1 + "----i2===" + i2);
//            return i1 - i2;
//        }, Flux.just(9, 1, 2, 6, 8), Flux.just(9, 4, 7, 5, 8)).subscribe(System.err::println);

        //Flux.mergeComparing((s1,s2)-> s1-s2, Flux.just(1), Flux.just(6), Flux.just(4), Flux.just(2), Flux.just(7)).subscribe(System.out::println);

        //Flux.zip(Flux.just(1, 2, 0), Flux.just(4, 5, 6,7)).subscribe(System.out::println);
        //可以通过bifunction来自定义压缩的方式,默认是放到一个数组中
        //Flux.zip(Flux.just(1, 2, 0), Flux.just(4, 5, 6, 7), (i1, i2) -> i1 + i2).subscribe(System.out::println);
        //所有的数据都必须符合断言返回true,否则返回false
        // Flux.just(1, 2, 0).all(integer -> integer > -1).subscribe(System.out::println);
        //任何一个数据符合断言就返回true,否则返回false
        //Flux.just(1, 2, 0).any(integer -> integer > 1).subscribe(System.out::println);
        //将数据通过as变成一个结果,具体变化由实际的业务掘金
//        String result = Flux.just(1, 2, 3).as(fi -> {
//            StringBuffer buffer = new StringBuffer();
//            fi.subscribe(integer -> {
//                buffer.append(integer);
//            });
//            return buffer.toString();
//        });
//       System.err.println(result);

        //  Flux.just(1, 2, 3,4,5,6,7,8).buffer(3).subscribe(System.out::println);
        //  Flux.just(1, 2, 3,4,5,6,7,8).buffer(3,1).subscribe(System.out::println);

//        Flux<Integer> cache = Flux.just(1, 2, 3, 4, 5).cache(2);
//        //第一次订阅是原始的所有的数据,然后会根据我们期望的cache的个数来将最后几个数据进行缓存
//        cache.subscribe(System.out::println);
//        //第二次次订阅后只有缓存的最后的个数的数据
//        cache.subscribe(System.out::println);
        //cache的时候可以设置缓存的有效期,在有效期之内我们第二次订阅开始都是缓存的数据,直到缓存过期,会重新输入原始的数据并再次做缓存,只要程序不结束,就一直循环往复
//        Flux<Integer> cache = Flux.just(1, 2, 3, 4, 5).cache(2, Duration.of(5, ChronoUnit.SECONDS));
//        cache.subscribe(System.out::println);
//        System.out.println("-----------------");
//        cache.subscribe(System.out::println);
//        System.out.println("-----------------");
//        cache.subscribe(System.out::println);
//        System.out.println("-----------------");
//        cache.subscribe(System.out::println);
//        System.out.println("-----------------");
//        Thread.sleep(6000);
//        cache.subscribe(System.out::println);
//        System.out.println("-----------------");
//        cache.subscribe(System.out::println);
//        System.out.println("-----------------");
//        cache.subscribe(System.out::println);
//        System.out.println("-----------------");
//        Thread.sleep(6000);
//        cache.subscribe(System.out::println);
//        System.out.println("-----------------");
//        cache.subscribe(System.out::println);
//        System.out.println("-----------------");
//        cache.subscribe(System.out::println);
//        System.out.println("-----------------");

        //cast是进行类型转换的,如果类型不对会出现类型转换异常
        // Flux.just(1, 2, 3, 4, 5).cast(String.class).subscribe(System.out::println);
//        Mono<Set<Integer>> mono = Flux.just(1, 2, 3, 4, 5).collect(Collectors.toSet());
//        mono.subscribe(System.err::println);

//        Flux.just(1, 2, 3, 4, 5, 6, 6).collect(
//                ()->"a",//将这个返回的a拼接到我们的每个数据的前面,这个只会执行一次
//                (t,u)->System.out.println(t+u)//t就是我们每次拼接到前面的那个a, u就是每次遍历的数据
//        ).subscribe(System.err::println);//订阅其实就是我们返回的那个a

        //Flux.just("def").defaultIfEmpty("abc").subscribe(System.err::println);
        //  Flux.just(1, 2, 3, 4, 5,5,6,5).distinct().subscribe(System.out::println);
        //distinct参数keyselector是用于判断当前数据是不是出现过,和返回值进行比较,只要没有和这个值不同的就保存,否则就认为是重复的数据
        // Flux.just(1, 2, 3, 4, 5,5,6,5).distinct(integer ->1).subscribe(System.out::println);
        // Flux.just(1, 2, 3, 4, 5,5,6,5).distinct(integer ->integer>4).subscribe(System.out::println);
        // Flux.just(1, 2, 3, 4, 5,5,4,4,4,4,4,4,4,4,5,6,5,7,4,5).distinctUntilChanged().subscribe(System.out::println);
//        Flux.just(1, 2, 3, 4, 0, 5, 6, 5)
//                .doOnNext(integer -> {System.err.println("当前遍历的数据是:" + integer);
//                    int i = 1 / integer;
//                })//在消费者的onNext之前先执行
//                .doOnComplete(() -> {
//                    System.err.println("doOnComplete 执行了--->");
//                })
//                .doOnError(throwable -> System.err.println("发生错误" ))
//                .subscribe(new Subscriber<Integer>() {
//                    private Subscription subscription;
//
//                    @Override
//                    public void onSubscribe(Subscription subscription) {
//                        this.subscription = subscription;
//                        subscription.request(1);
//                    }
//
//                    @Override
//                    public void onNext(Integer integer) {
//                        System.err.println("FluxTest中的方法onNext 执行了--->" + integer);
//                        subscription.request(1);
//                    }
//
//                    /**
//                     * 我们在生产数据的时候或者在到达订阅者之前出现的异常会到这里
//                     * @param throwable
//                     */
//                    @Override
//                    public void onError(Throwable throwable) {
//                        System.err.println("FluxTest中的方法onError 执行了--->");
//                    }
//
//                    @Override
//                    public void onComplete() {
//                        System.err.println("FluxTest中的方法onComplete 执行了--->");
//                    }
//                });
        //Flux.just(1, 2, 3, 4, 5,5,6,5).filter(integer -> integer>4).subscribe(System.out::println);
        //Flux.just(1, 2, 3, 4, 5,5,6,5).map(integer -> integer+10).subscribe(System.out::println);
        // Flux.just(1, 2, 3, 4, 5,5,6,5).map(integer -> new User(integer)).subscribe(System.out::println);
//        Flux.just(1, 2, 3, 4, 8, 5, 6, 6, 10, 100).doOnNext(integer -> {
//            System.err.println("当前拿到了" + integer);
//        }).doOnComplete(()->{
//            System.err.println("结束了");
//        }).then(Mono.just("abc")).subscribe(System.err::println);

        countDownLatch.await();
    }
}
